/*
 * MessageInputStream.java   0.0.1    28/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */ 
package org.mentalsmash.tazio.net.cs;

import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.io.StreamCorruptedException;
import java.nio.ByteBuffer;

import org.mentalsmash.tazio.net.cs.protocol.MessageHeader;
import org.mentalsmash.tazio.net.cs.protocol.NIOObjectContainer;


/**
 * This {@link java.io.InputStream} implementation can be used to read an 
 * object serialized using the protocol defined by the {@link NIOObjectContainer}
 * class, from an underlying stream.
 * 
 * @author Andrea Reale <andrea.reale@gmail.com
 * @version 0.0.1
 * @see NIOObjectContainer
 * @see MessageOutputStream
 */
public class MessageInputStream extends InputStream {
    
    private final InputStream inputStream;
    
    /**
     * Create a new <tt>MessageInputStream</tt> using the given {@link InputStream}
     * as the underlying stream.
     * @param input an input stream to read data from
     */
    public MessageInputStream(InputStream input) {
	inputStream = input;
    }

    /**
     * Calls <code>read()</code> on the underlying stream.
     * @throws IOException if an I/O error occurs.
     * @see InputStream#read()
     */
    @Override
    public int read() throws IOException {
	return inputStream.read();
    }
    
    /**
     * Calls <code>read(byte,int,int)</code> ont he underlying stream.
     * @throws IOException if an I/O error occurs.
     * @see InputStream#read(byte[], int, int)
     */
    @Override
    public int read(byte[] b,int off,int len) throws IOException {
	return inputStream.read(b, off, len);
    }
    
    /**
     * Calls <code>available()</code> on the underlying stream.
     * @throws IOException if an I/O error occurs.
     * @see InputStream#available()
     */
    @Override
    public int available() throws IOException {
	return inputStream.available();
    }
    
    /**
     * Calls <code>skip()</code> on the underlying stream.
     * @throws IOException if an I/O error occurs.
     * @see InputStream#skip(long)
     */
    @Override
    public long skip(long n) throws IOException {
	return inputStream.skip(n);
    }
    
    /**
     * Calls <code>close()</code> on the underlying stream.
     * @throws IOException if an I/O error occurs.
     * @see InputStream#close()
     */
    @Override
    public void close() throws IOException {
	inputStream.close();
    }
    
    /**
     * Tries to read a complete message from the underlying stream, and to retrieve
     * the serializable message itself using the protocol defined by the
     * {@link NIOObjectContainer} class.
     * <p>
     * If the underlying stream does not contain a valid message it throws a 
     * {@link StreamCorruptedException}.
     * </p>
     * <p>
     * If the class of the Serializable object to retrieve cannot be load by the 
     * current <i>Classloader</i> throws a {@link ClassNotFoundException}.
     * </p>
     * @return the retrieved message
     * @throws IOException if an I/O error occurs.
     * @throws ClassNotFoundException if the class of the message cannot be found
     *                                by the current Classloader
     */
    public Serializable readMessage() throws IOException, ClassNotFoundException {
	byte[] headerBytes = consumeHeader();
	MessageHeader header = MessageHeader.readHeader(headerBytes);
	int payloadLength = header.getPayloadLength();
	
	byte[] payloadBytes = consumePayload(payloadLength);
	ByteBuffer completeMessage = ByteBuffer.allocate(headerBytes.length + payloadBytes.length);
	completeMessage.put(headerBytes).put(payloadBytes);
	
	return unmarshal(completeMessage.array());
    }
    
    /**
     * Tries to read <code>length</code> bytes from the underlying stream
     * @param length the number of bytes to read
     * @return the bytes read
     * @throws IOException if an I/O error occurs.      
     */
    private byte[] consumePayload(int length) throws IOException {
	int nread =0;
	byte[] payloadBytes = new byte[length];
	int remainingToRead = length;
	
	while(remainingToRead != 0) {
	    int read = inputStream.read(payloadBytes,nread,remainingToRead);
	    remainingToRead -= read;
	    nread += read;
	}
	
	return payloadBytes;
    }
    


    /**
     * Retrieves the serialized payload from a complete message (header+payload)
     * using the protocol defined by the {@link NIOObjectContainer} class
     * @param encapsulated the marshalled message (header+payload) byte array
     * @return the unmarshalled payload of the message
     * @throws IOException if an I/O error occurs.
     * @throws ClassNotFoundException If the class of the payload cannot be found
     *                                by the current Classloader
     */
    private Serializable unmarshal(byte[] encapsulated) throws IOException, 
    	ClassNotFoundException {
	
	NIOObjectContainer container = null;

	container =
	    NIOObjectContainer.readContainer(encapsulated);

	
	return  container.getPayload();
    }
    
    /**
     * Tries to read {@link MessageHeader#HEADER_SIZE} bytes from the
     * underlying stream
     * @return the data read
     * @throws IOException if an I/O error occurs.
     */
    private byte[] consumeHeader() throws IOException {
	int nread = 0;
	byte[] headerBytes = new byte[MessageHeader.HEADER_SIZE];
	int remainingToRead = MessageHeader.HEADER_SIZE;
	
	while(remainingToRead != 0) {
	    int read = inputStream.read(headerBytes,nread,remainingToRead);
	    remainingToRead -= read;
	    nread += read;
	}
	
	return headerBytes;

    }

}
